package cn.dfun.sample.flink.apitest
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.runtime.state.memory.MemoryStateBackend
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
  * process function使用侧输出流分流
  */
object SideOutputTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 该方式弃用,推荐使用静态方法
//    env.setStateBackend(new FsStateBackend())
    // RocksDBStateBackend需要引入依赖
    // 可以设置增量存盘参数
//    env.setStateBackend(new RocksDBStateBackend())
    val inputStream = env.socketTextStream("node-01", 7777)

    // 包装成样例类
    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
    val highTempStream = dataStream
        .process(new SplitTempProcessor(30))
    highTempStream.print()
    highTempStream.getSideOutput(new OutputTag[(String, Long, Double)]("low")).print("low")
    env.execute("side output test")
  }
}

class SplitTempProcessor(threshold: Double) extends ProcessFunction[SensorReading, SensorReading] {
  override def processElement(i: SensorReading, context: ProcessFunction[SensorReading, SensorReading]#Context, collector: Collector[SensorReading]): Unit = {
    // 如果当前温度值大于30,输出到主流
    if(i.temperature > threshold) {
      collector.collect(i)
    } else {
      context.output(new OutputTag[(String, Long, Double)]("low"), (i.id, i.timestamp, i.temperature))
    }
  }
}
